package executor

import (
	"bufio"
	ckeadm "ckeadm/cmd/apis"
	ckeadmv1beta1 "ckeadm/cmd/apis/v1alpha1"
	logscmd "ckeadm/cmd/logs"
	"ckeadm/cmd/runner"
	"ckeadm/constants"
	"ckeadm/util"
	"ckeadm/util/runtime"
	"container/list"
	"encoding/base64"
	"fmt"
	"github.com/sirupsen/logrus"
	"io"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	utilsexec "k8s.io/utils/exec"
	"math/rand"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

//容器中额外挂载一个/script
func CreateComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig) error {
	//调用对应节点的ckectl命令执行安装
	//docker需要传安装包和脚本
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			logrus.Infof("copy the script of component %s from image: %s", component.Name, GetImageName(cluster, component))
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		//copy script to /opt/script-template
		for _, component := range *components {
			// #nosec
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}

	if len(*components) > 1 {
		componentsMap := make(map[string]ckeadmv1beta1.CkeComponent)
		_, ckecomponentsSorting := util.CheckAndSort(*components)
		logrus.Infof("get ckecomponentsSorting: %+v", ckecomponentsSorting)
		for _, component := range *components {
			componentsMap[component.Name] = component
		}
		ch := make(chan NodeResult)
		clusterExecuted := false
		suffix := RandString(6)
		for _, node := range *ckeNodes {
			//选择一个master执行cluster操作
			isClusterExecutedNode := false
			if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
				clusterExecuted = true
				isClusterExecutedNode = true
			}
			go CreateComponentsInNode(isClusterExecutedNode, node, cluster, keyFile, watchConfig, ckeNodes, ch, ckecomponentsSorting, componentsMap, suffix)
		}

		routineCount := len(*ckeNodes)
		allSuccess := true
		nodesResults := make([]NodeResult, 0)
		for {
			select {
			case res := <-ch:
				routineCount--
				if res.nodeName != "" {
					//record failed components on node xxx
					RecordStatus(cluster.Name, res.nodeName, constants.CkeStatusFailed, -1, strings.Split(res.componentName, ","))
					allSuccess = false
					nodesResults = append(nodesResults, res)
				}
				if routineCount == 0 {
					RemoveComponentScripts()
					if !allSuccess {
						for _, r := range nodesResults {
							logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
						}
						RecordEndTime(cluster.Name)
						return fmt.Errorf("not all nodes are successful, see log for detail")
					} else {
						RecordEndTime(cluster.Name)
						logrus.Info("All nodes are successful.")
						return nil
					}
				}
			}
		}
	} else {
		//	集群部署成功之后，单独部署组件时，在job类的pod中执行以下逻辑
		//运行时安装 需要直传安装包和脚本
		//确定执行节点，确定节点的角色 调用ckectl
		//cluster级别的直接起pod执行
		ch := make(chan NodeResult)
		clusterExecuted := false
		suffix := RandString(6)
		matchNodes, notEnough := getMatchNodes(ckeNodes, (*components)[0].Spec.Nodes, (*components)[0].Spec.NodeSelector, (*components)[0].Spec.Replicas)
		if notEnough {
			return fmt.Errorf("matchNodes %s are not enough for the replicas(%v) of component %s ", strings.Join(matchNodes, ","), (*components)[0].Spec.Replicas, (*components)[0].Name)
		}
		for _, node := range *ckeNodes {
			//选择一个master执行cluster操作
			isClusterExecutedNode := false
			if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
				clusterExecuted = true
				isClusterExecutedNode = true
			}
			go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
				keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string, matchNodes []string) {
				for _, component := range *components {
					logrus.Infof("Creating component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
					workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/create"
					if !util.Exists(workPath) {
						logrus.Errorf("create dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
						ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
						return
					}
					dirs, err := util.ListSubDirsInPath(workPath)
					if err != nil {
						logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
						ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
						return
					}
					sort.Sort(sort.StringSlice(dirs))

					//去除序号
					dirWithoutNumber := make([]string, 0)
					for _, dir := range dirs {
						reg := regexp.MustCompile(`^[0-9]+-`)
						dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
					}

					//过滤出componentCr中指定的节点
					match := false
					if util.ContainsStrings(matchNodes, node.Name) {
						match = true
					}

					//dir: master slave node cluster
					for index, dir := range dirWithoutNumber {
						//获取节点 调用
						if match && isCluster && (dir == "cluster") {
							//todo 改造成job形式
							err = ExecuteCkectlCreate(component, node, cluster, allNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
							if err != nil {
								logrus.Errorf("ExecuteCkectlCreate failed at node %s: %s", node.ObjectMeta.Name, err.Error())
								ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
								return
							}
							continue
						}
						if match && NodeIsInDir(dir, node.Spec.Roles) {
							//调用指定节点的ckectl component create
							err = ExecuteCkectlCreate(component, node, cluster, allNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
							if err != nil {
								logrus.Errorf("ExecuteCkectlCreate failed at node %s: %s", node.ObjectMeta.Name, err.Error())
								ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
								return
							}
						}
					}

					//	记录组件安装结果
					p := filepath.Join(constants.CkeCompInstalled, cluster.Name)
					var recordStatusCmd = fmt.Sprintf("mkdir -p %s && echo installed > %s", p, filepath.Join(p, component.Name + "-install.log"))
					_, _, _, _ = util.SetupSSH(node.Spec.Address, keyFile).Run(recordStatusCmd, 60*time.Second)
				}
				ch <- NodeResult{}
				return
			}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix, matchNodes)
		}

		routineCount := len(*ckeNodes)
		allSuccess := true
		nodesResults := make([]NodeResult, 0)
		for {
			select {
			case res := <-ch:
				routineCount--
				if res.nodeName != "" {
					allSuccess = false
					nodesResults = append(nodesResults, res)
				}
				if routineCount == 0 {
					RemoveComponentScripts()
					if !allSuccess {
						for _, r := range nodesResults {
							logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
						}
						return fmt.Errorf("not all nodes are successful, see log for detail")
					} else {
						logrus.Info("All nodes are successful.")
						return nil
					}
				}
			}
		}
	}
}

type NodeResult struct {
	nodeName      string
	componentName string
}

func getMatchNodes(ckenodes *[]ckeadmv1beta1.CkeNode, nodes, nodeSelector []string, replicas int) ([]string, bool) {
	notEnough := false
	matchNodes := make([]string, 0)
	if len(nodes) > 0 {
		matchNodes = nodes
		return matchNodes, notEnough
	} else if len(nodeSelector) > 0 {
		for _, ckenode := range *ckenodes {
			if util.ContainsStrings(ckenode.Spec.Labels, nodeSelector...) {
				matchNodes = append(matchNodes, ckenode.Name)
			}
		}
	} else {
		for _, ckenode := range *ckenodes {
			matchNodes = append(matchNodes, ckenode.Name)
		}
	}
	sort.Strings(matchNodes)
	if replicas != 0 {
		if replicas <= len(matchNodes) {
			matchNodes = matchNodes[:replicas]
		} else {
			notEnough = true
			return matchNodes, notEnough
		}
	}
	return matchNodes, notEnough
}

func CreateComponentsInNode(isCluster bool, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult,
	ckecomponentsSorting [][]string, componentsMap map[string]ckeadmv1beta1.CkeComponent, suffix string) {
	allSuccess := true
	failedComponents := make([]string, 0)

	for index, activeComponents := range ckecomponentsSorting {
		//record activeComponents are installing on node xxx, progress xxx%
		RecordStatus(cluster.Name, node.Name, constants.CkeStatusInstall, getProgress(ckecomponentsSorting, index), activeComponents)
		var wg sync.WaitGroup
		for _, activeComponent := range activeComponents {
			wg.Add(1)
			//	一个协程负责一个组件的安装
			go func(isCluster bool, node *ckeadmv1beta1.CkeNode, cluster *ckeadmv1beta1.CkeCluster,
				keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ckecomponentsSorting [][]string,
				componentsMap map[string]ckeadmv1beta1.CkeComponent, component ckeadmv1beta1.CkeComponent, suffix string, allSuccess *bool, failedComponents *[]string) {
				defer func() {
					wg.Done()
				}()

				if component.Annotations != nil {
					if val, ok := component.Annotations[constants.ForceInstall];ok {
						logrus.Infof("Node:%s,Component:%s,Annotations:%s",node.Name,component.Name,val)
					}
				}


				// 检查是否有强制安装的注解，如果有则安装或重装，否则检查是否有安装结果记录;如果有则跳过安装;
				if component.Annotations != nil {
					if val, ok := component.Annotations[constants.ForceInstall]; !ok || val != "true" {
						//	check whether component already installed successfully
						var lsCmd = fmt.Sprintf("ls %s", filepath.Join(constants.CkeCompInstalled, cluster.Name, component.Name + "-install.log"))
						stdout, stderr, _, err := util.SetupSSH(node.Spec.Address, keyFile).Run(lsCmd, 60*time.Second)
						logrus.Infof("check whether component already installed,Node:%s,Component:%s,stdout:%s,stderr:%s,err:%v",node.Name,component.Name,stdout,stderr,err)

						if err != nil && !strings.Contains(stderr, "No such file or directory") && !strings.Contains(stderr, "没有那个文件或目录") {
							logrus.Errorf("check whether component already installed,Node:%s,Component:%s,failed,error: %s", node.Name, component.Name, stdout + stderr + err.Error())
							return
						}
						if strings.Contains(stdout, filepath.Join(constants.CkeCompInstalled, cluster.Name, component.Name + "-install.log")) {
							//logrus.Infof("component %s at %s already installed,skip installing.\n", component.Name, node.Name)
							logrus.Infof("Skip Installing Component,Node:%s,Component:%s\n", node.Name, component.Name)
							return
						}
					}
				} else {
					var lsCmd = fmt.Sprintf("ls %s", filepath.Join(constants.CkeCompInstalled, cluster.Name, component.Name + "-install.log"))
					stdout, stderr, _, err := util.SetupSSH(node.Spec.Address, keyFile).Run(lsCmd, 60*time.Second)
					logrus.Infof("Node:%s,Component:%s,stdout:%s,stderr:%s,err:%v",node.Name,component.Name,stdout,stderr,err)

					if err != nil && !strings.Contains(stderr, "No such file or directory") && !strings.Contains(stderr, "没有那个文件或目录") {
						logrus.Errorf("check '%s' whether already installed at %s : failed,error: %s", component.Name, node.Name, stdout + stderr + err.Error())
						return
					}
					if strings.Contains(stdout, filepath.Join(constants.CkeCompInstalled, cluster.Name, component.Name + "-install.log")) {
						//logrus.Infof("component %s at %s already installed,skip installing.\n", component.Name, node.Name)
						logrus.Infof("Skip Installing Component,Node:%s,Component:%s\n", node.Name, component.Name)
						return
					}
				}

				logrus.Infof("Creating component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/create"
				if !util.Exists(workPath) {
					logrus.Errorf("create dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					*allSuccess = false
					*failedComponents = append(*failedComponents, component.Name)
					return
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					*allSuccess = false
					*failedComponents = append(*failedComponents, component.Name)
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				//过滤出componentCr中指定的节点
				match := false
				if util.ContainsStrings(component.Status.Nodes, node.Name) {
					match = true
				}

				execCout := 0
				//dir: master slave node cluster
				for index, dir := range dirWithoutNumber {
					//获取节点 调用
					if match && isCluster && (dir == "cluster") {
						err = ExecuteCkectlCreate(component, *node, *cluster, allNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlCreate for component %s failed at node %s: %s", component.Name, node.ObjectMeta.Name, err.Error())
							*allSuccess = false
							*failedComponents = append(*failedComponents, component.Name)
							return
						}
						execCout ++
						continue
					}
					if match && NodeIsInDir(dir, node.Spec.Roles) {
						//调用指定节点的ckectl component create
						err = ExecuteCkectlCreate(component, *node, *cluster, allNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlCreate for component %s failed at node %s: %s", component.Name, node.ObjectMeta.Name, err.Error())
							*allSuccess = false
							*failedComponents = append(*failedComponents, component.Name)
							return
						}
						execCout ++
					}
				}

				//	记录组件安装结果
				p := filepath.Join(constants.CkeCompInstalled, cluster.Name)
				var recordStatusCmd = fmt.Sprintf("mkdir -p %s && echo installed > %s", p, filepath.Join(p, component.Name + "-install.log"))
				_, _, _, _ = util.SetupSSH(node.Spec.Address, keyFile).Run(recordStatusCmd, 60*time.Second)
				return
			}(isCluster, &node, &cluster, keyFile, watchConfig, allNodes, ckecomponentsSorting, componentsMap, componentsMap[activeComponent], suffix, &allSuccess, &failedComponents)
		}

		wg.Wait()

		if !allSuccess {
			ch <- NodeResult{nodeName: node.Name, componentName: strings.Join(failedComponents, ",")}
			return
		} else {
			logrus.Infof("Components %s install in node %s successful.", strings.Join(activeComponents, ","), node.Name)
		}
	}
	//record success on one node
	RecordStatus(cluster.Name, node.Name, constants.CkeStatusCompleted, 100, nil)
	ch <- NodeResult{}
	return
}

//execute ckectl cmd at one node to process one component
func ExecuteCkectlCreate(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {

	//logrus.Infof(">>>>>>>>Node:%s,Component:%s,Dir:%s",node.Name,component.Name,dir)

	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}

	if component.Labels["containerRuntime"] == "true" {
		var mkdirCmd = fmt.Sprintf("mkdir -p %s", constants.CkectlScriptPath)
		_, _, _, err := util.SetupSSH(node.Spec.Address, keyFile).Run(mkdirCmd, 60*time.Second)
		if err != nil {
			logrus.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
			return fmt.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
		}

		scpCmd := fmt.Sprintf("scp -o StrictHostKeyChecking=no -i %s -r %s %s:%s", keyFile, constants.CkeComponentScriptPath+"/"+component.Name, node.Spec.Address, constants.CkectlScriptPath)
		// #nosec
		cmd := exec.Command("sh", "-c", scpCmd)
		output, err := cmd.CombinedOutput()
		logrus.Infof("scp output: %s", string(output))
		if err != nil {
			logrus.Errorf("execute scp cmd failed: %s", err.Error())
			return fmt.Errorf("execute scp cmd failed: %s", err.Error())
		}

		var ckectlCmd = fmt.Sprintf(`%s component create --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component create failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop1:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop1
				case outline := <-stdoutChan:
					logrus.Info("ckectlCmd:stdout:" + node.Name + "_" + component.Name + ":", outline)
				case errline := <-stderrChan:
					logrus.Info("ckectlCmd:stderr:" + node.Name + "_" + component.Name + ":", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl component create failed, err: command timeout")
				return fmt.Errorf("ckectl component create failed, err: command timeout")
			}
			logrus.Infof("Execute Script Successful,Node:%s,Component:%s,Dir:%s",node.Name,component.Name,dir)
		}
	} else {
		var ckectlCmd = fmt.Sprintf(`%s component create --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component create failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop2:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop2
				case outline := <-stdoutChan:
					logrus.Info("ckectlCmd:stdout:" + node.Name + "_" + component.Name + ":", outline)
				case errline := <-stderrChan:
					logrus.Info("ckectlCmd:stderr:" + node.Name + "_" + component.Name + ":", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl component create failed, err: command timeout")
				return fmt.Errorf("ckectl component create failed, err: command timeout")
			}
			logrus.Infof("Execute Script Successful,Node:%s,Component:%s,Dir:%s",node.Name,component.Name,dir)
		}
	}
	return nil
}

//extract script into sub dir
func ExtractScript(containerName, imageName string, component ckeadmv1beta1.CkeComponent) error {
	dockerRunTime, err := runtime.NewContainerRuntime(utilsexec.New())
	utilruntime.HandleError(err)

	if exist, err := dockerRunTime.ImageExists(imageName); err != nil || !exist {
		err = dockerRunTime.PullImage(imageName)
		if err != nil {
			return fmt.Errorf("fail pull images: %s, err: %s", imageName, err.Error())
		}
	}

	//清空对应插件子目录
	_ = os.RemoveAll(constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name)

	err = dockerRunTime.RemoveContainers([]string{containerName})
	dockerArgs := []string{
		"run",
		"--rm",
		"-v", constants.CkeComponentScriptPath + ":" + constants.CkeComponentScriptPath,
		"--name", containerName,
	}

	// #nosec
	var running = exec.Command("docker", append(dockerArgs, imageName, "/bin/cp", "-rf", "/template", constants.CkeComponentScriptPath+"/"+component.ObjectMeta.Name)...)

	_, err = running.CombinedOutput()
	return err
}

func GetImageName(cluster ckeadmv1beta1.CkeCluster, component ckeadmv1beta1.CkeComponent) string {
	return runner.GetImage(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port, constants.ScriptImageRepo+component.ObjectMeta.Name,
		GetScriptVersion(component), cluster.Spec.ClusterParameters.ImageManifestEnabled)
}

func GetScriptVersion(component ckeadmv1beta1.CkeComponent) string {
	if component.Spec.ImageTag != "" {
		return component.Spec.ImageTag
	} else {
		return component.Spec.Version
	}
}

func InArray(obj string, array []string) bool {
	for _, item := range array {
		if obj == item {
			return true
		}
	}
	return false
}

//nodeGroup master
//nodeGroup master[and]worker[or]etcd[not]ntp
//master worker and  etcd  or  ntp not
//递归 转后缀表达式
//判断目录名是否匹配节点角色，并且处理扩容缩容情况
func NodeIsInDir(nodeGroup string, roles []string) bool {
	//kube-scale
	if nodeGroup == constants.ScaleUpDir {
		for _, role := range roles {
			if strings.Contains(role, "kube-scale") {
				return true
			}
		}
	}
	//kube-scaledown
	if nodeGroup == constants.ScaleDownDir {
		for _, role := range roles {
			if strings.Contains(role, "kube-scaledown") {
				return true
			}
		}
	}
	op := []string{"and", "or", "not"}
	array := regexp.MustCompile(`[\[\]]+`).Split(nodeGroup, -1)
	//转换为后缀表达式
	for i := 0; i < len(array); i++ {
		if InArray(array[i], op) {
			array[i], array[i+1] = array[i+1], array[i]
			i++
		}
	}
	//convert to true false
	for index, item := range array {
		if !InArray(item, op) {
			if InArray(item, roles) {
				array[index] = "true"
			} else {
				array[index] = "false"
			}
		}
	}

	if len(array) == 1 {
		if array[0] == "true" {
			return true
		} else {
			return false
		}
	}

	stack := list.New()
	for _, item := range array {
		if !InArray(item, op) {
			stack.PushBack(item)
		} else {
			b := stack.Back().Value
			stack.Remove(stack.Back())
			a := stack.Back().Value
			stack.Remove(stack.Back())
			result := Calc(a.(string), b.(string), item)
			stack.PushBack(result)
		}
	}
	if stack.Back().Value.(string) == "true" {
		return true
	} else {
		return false
	}
}

func Calc(a, b, op string) string {
	switch op {
	case "and":
		if (a == "true") && (b == "true") {
			return "true"
		} else {
			return "false"
		}
	case "or":
		if (a == "true") || (b == "true") {
			return "true"
		} else {
			return "false"
		}
	case "not":
		if (a == "true") && (b == "false") {
			return "true"
		} else {
			return "false"
		}
	default:
		return "false"
	}
}

//扩容多个组件
//容器中额外挂载一个/script
func ScaleUpComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig, addNode string) error {
	//调用对应节点的ckectl命令执行安装
	//docker需要传安装包和脚本
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		//copy script to /opt/script-template
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		for _, component := range *components {
			// #nosec
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}

	componentsMatchNodesMap := make(map[string][]string)
	for _, component := range *components {
		matchNodes := util.AppendIfNotExist(component.Status.Nodes, addNode)
		componentsMatchNodesMap[component.Name] = matchNodes
	}

	//cluster级别的直接起pod执行
	//需要扩容的节点从pending节点列表中获取，脚本负责根据这里判断是否是扩容节点
	//其余已经安装的节点从status的nodes列表中获取
	//每个组件 发送到该组件的所有已经安装节点和待扩容节点执行操作
	//不需要额外计算目标节点
	type NodeResult struct {
		nodeName      string
		componentName string
	}
	ch := make(chan NodeResult)
	clusterExecuted := false
	suffix := RandString(6)
	for _, node := range *ckeNodes {
		//选择一个master执行cluster操作
		isClusterExecutedNode := false
		if (!clusterExecuted) && InArray("master", node.Spec.Roles) { //  util.DialPortTimeout(node.Spec.Address, constants.SSHConnPort)
			clusterExecuted = true
			isClusterExecutedNode = true
		}
		go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
			keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string, componentsMatchNodesMap map[string][]string) {
			for _, component := range *components {
				logrus.Infof("Scaling up component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/scaleup"
				if !util.Exists(workPath) {
					logrus.Infof("scaleup dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					continue
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				match := false
				if util.ContainsStrings(componentsMatchNodesMap[component.Name], node.Name) {
					match = true
				}

				//dir: master slave node cluster
				//获取节点类别的执行顺序
				for index, dirname := range dirWithoutNumber {
					// 节点 role 匹配到 dirname 则执行 ( role == dirname OR master[0] == cluster )
					if match && (NodeIsInDir(dirname, node.Spec.Roles) || isCluster && (dirname == constants.ClusterDir)) {
						//调用指定节点的ckectl component scaleup
						err = ExecuteCkectlScaleUp(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							// 除 cluster 执行节点和 kube-scale 节点外，忽略其它节点执行错误 (保证有坏节点仍能正常扩容)
							if !util.ContainsStrings(node.Spec.Roles, constants.ScaleUpDir) && !isCluster {
								logrus.Warningf("ExecuteCkectlScaleUp failed at node %s: %s, ignoring ...", node.ObjectMeta.Name, err.Error())
								ch <- NodeResult{}
							} else {
								logrus.Errorf("ExecuteCkectlScaleUp failed at node %s: %s", node.ObjectMeta.Name, err.Error())
								ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							}
							return
						}
					}
				}
			}
			ch <- NodeResult{}
			return
		}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix, componentsMatchNodesMap)
	}

	routineCount := len(*ckeNodes)
	allSuccess := true
	nodesResults := make([]NodeResult, 0)
	for {
		select {
		case res := <-ch:
			routineCount--
			if res.nodeName != "" {
				allSuccess = false
				nodesResults = append(nodesResults, res)
			}
			if routineCount == 0 {
				RemoveComponentScripts()
				if !allSuccess {
					for _, r := range nodesResults {
						logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
					}
					return fmt.Errorf("not all nodes are successful, see log for detail")
				} else {
					logrus.Info("All nodes are successful.")
					return nil
				}
			}
		}
	}
}

//execute ckectl scaleup cmd at one node to process one component
func ExecuteCkectlScaleUp(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {
	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}
	if component.Labels["containerRuntime"] == "true" {
		var mkdirCmd = fmt.Sprintf("mkdir -p %s", constants.CkectlScriptPath)
		_, _, _, err := util.SetupSSH(node.Spec.Address, keyFile).Run(mkdirCmd, 60*time.Second)
		if err != nil {
			logrus.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
			return fmt.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
		}

		scpCmd := fmt.Sprintf("scp -o StrictHostKeyChecking=no -i %s -r %s %s:%s", keyFile, constants.CkeComponentScriptPath+"/"+component.Name, node.Spec.Address, constants.CkectlScriptPath)
		// #nosec
		cmd := exec.Command("sh", "-c", scpCmd)
		output, err := cmd.CombinedOutput()
		logrus.Infof("scp output: %s", string(output))
		if err != nil {
			logrus.Errorf("execute scp cmd failed: %s", err.Error())
			return fmt.Errorf("execute scp cmd failed: %s", err.Error())
		}

		var ckectlCmd = fmt.Sprintf(`%s component scaleup --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component scaleup failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop1:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop1
				case outline := <-stdoutChan:
					logrus.Info("out:", outline)
				case errline := <-stderrChan:
					logrus.Info("out:", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl component scaleup failed, err: command timeout")
				return fmt.Errorf("ckectl component scaleup failed, err: command timeout")
			}
		}
		return nil
	} else {

		var ckectlCmd = fmt.Sprintf(`%s component scaleup --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component scaleup failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop2:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop2
				case outline := <-stdoutChan:
					logrus.Info("out:", outline)
				case errline := <-stderrChan:
					logrus.Info("out:", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl component scaleup failed, err: command timeout")
				return fmt.Errorf("ckectl component scaleup failed, err: command timeout")
			}
		}
		return nil
	}
}

//缩容多个组件
//容器中额外挂载一个/script
func ScaleDownComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig) error {
	//调用对应节点的ckectl命令执行安装
	//docker需要传安装包和脚本
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		//copy script to /opt/script-template
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		for _, component := range *components {
			// #nosec
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}
	//cluster级别的直接起pod执行
	//每个组件 发送到该组件的所有已经安装节点节点执行操作
	//不需要额外计算目标节点
	type NodeResult struct {
		nodeName      string
		componentName string
	}
	ch := make(chan NodeResult)
	clusterExecuted := false
	suffix := RandString(6)
	for _, node := range *ckeNodes {
		//选择一个master执行cluster操作
		isClusterExecutedNode := false
		if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
			clusterExecuted = true
			isClusterExecutedNode = true
		}
		go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
			keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string) {
			for _, component := range *components {
				logrus.Infof("Scaling down component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/scaledown"
				if !util.Exists(workPath) {
					logrus.Infof("scaledown dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					continue
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				installedNodesName := component.Status.Nodes
				//dir: master slave node cluster
				//获取节点类别的执行顺序
				for index, dir := range dirWithoutNumber {
					//获取节点 调用
					if InArray(node.ObjectMeta.Name, installedNodesName) && isCluster && (dir == "cluster") {
						//todo 改造成job形式
						err = ExecuteCkectlScaleDown(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlScaleDown failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
						continue
					}
					if InArray(node.ObjectMeta.Name, installedNodesName) && NodeIsInDir(dir, node.Spec.Roles) {
						//调用指定节点的ckectl component scaledown
						err = ExecuteCkectlScaleDown(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlScaleDown failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
					}
				}
			}
			ch <- NodeResult{}
			return
		}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix)
	}

	routineCount := len(*ckeNodes)
	allSuccess := true
	nodesResults := make([]NodeResult, 0)
	for {
		select {
		case res := <-ch:
			routineCount--
			if res.nodeName != "" {
				allSuccess = false
				nodesResults = append(nodesResults, res)
			}
			if routineCount == 0 {
				RemoveComponentScripts()
				if !allSuccess {
					for _, r := range nodesResults {
						logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
					}
					return fmt.Errorf("not all nodes are successful, see log for detail")
				} else {
					logrus.Info("All nodes are successful.")
					return nil
				}
			}
		}
	}
}

//execute ckectl scaledown cmd at one node to process one component
func ExecuteCkectlScaleDown(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {
	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}

	var ckectlCmd = fmt.Sprintf(`%s component scaledown --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
		constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
	//日志暂未加任何包装
	stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
	if err != nil {
		logrus.Errorf("ckectl component scaledown failed, err: %s", err.Error())
		return err
	} else {
		// read from the output channel until the done signal is passed
		isTimeout := true
	loop2:
		for {
			select {
			case isTimeout = <-doneChan:
				break loop2
			case outline := <-stdoutChan:
				logrus.Info("out:", outline)
			case errline := <-stderrChan:
				logrus.Info("out:", errline)
			case err = <-errChan:
			}
		}

		// get exit code or command error.
		if err != nil {
			fmt.Println("err: " + err.Error())
			return err
		}

		// command time out
		if !isTimeout {
			logrus.Errorf("ckectl component scaledown failed, err: command timeout")
			return fmt.Errorf("ckectl component scaledown failed, err: command timeout")
		}
	}

	return nil
}

//删除多个组件
//容器中额外挂载一个/script
func DeleteComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig) error {
	//调用对应节点的ckectl命令执行安装
	//docker需要传安装包和脚本
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			logrus.Infof("copy the script of component %s from image: %s", component.Name, GetImageName(cluster, component))
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		//copy script to /opt/script-template
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		for _, component := range *components {
			// #nosec
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}
	//cluster级别的直接起pod执行
	//每个组件 发送到该组件的所有已经安装节点节点执行操作
	//不需要额外计算目标节点
	type NodeResult struct {
		nodeName      string
		componentName string
	}
	ch := make(chan NodeResult)
	clusterExecuted := false
	suffix := RandString(6)
	for _, node := range *ckeNodes {
		//选择一个master执行cluster操作
		isClusterExecutedNode := false
		if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
			clusterExecuted = true
			isClusterExecutedNode = true
		}
		go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
			keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string) {
			for _, component := range *components {
				//	检查是否有跳过删除的注解，有则跳过；反之亦然
				if component.Annotations != nil {
					if val, ok := component.Annotations[constants.SkipDelete]; ok && val == "true" {
						logrus.Infof("skip uninstall...,because of specified skipDelete,Node:%s,Component:%s", node.Name, component.ObjectMeta.Name)
						continue
					}
				}
				//	删除标志文件（记录安装相关）
				var deleteStatusCmd = fmt.Sprintf("rm  %s", filepath.Join(constants.CkeCompInstalled, cluster.Name, component.Name + "-install.log"))
				_, stderr, _, err := util.SetupSSH(node.Spec.Address, keyFile).Run(deleteStatusCmd, 60*time.Second)
				//logrus.Infof("Node:%s,Component:%s,stdout:%s,stderr:%s,isTimeout:%t,err:%v",node.Name,component.Name,stdout,stderr,isTimeout,err)

				if strings.Contains(stderr,"No such file or directory") || strings.Contains(stderr,"没有那个文件或目录") {
					//	防止卸载过程中出现异常，当卸载没有安装的组件时，跳过执行组件对应的删除脚本
					logrus.Infof("component not installed, skip uninstall...,Node:%s,Component:%s", node.Name, component.ObjectMeta.Name)
					continue
				}
				logrus.Infof("Deleting component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/delete"
				if !util.Exists(workPath) {
					logrus.Infof("delete dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					continue
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				installedNodesName := component.Status.Nodes
				//dir: master slave node cluster
				//获取节点类别的执行顺序
				for index, dir := range dirWithoutNumber {
					//获取节点 调用
					if InArray(node.ObjectMeta.Name, installedNodesName) && isCluster && (dir == "cluster") {
						//todo 改造成job形式
						err = ExecuteCkectlDelete(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlDelete failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
						continue
					}
					if InArray(node.ObjectMeta.Name, installedNodesName) && NodeIsInDir(dir, node.Spec.Roles) {
						//调用指定节点的ckectl component delete
						err = ExecuteCkectlDelete(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlDelete failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
					}
				}
			}
			ch <- NodeResult{}
			return
		}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix)
	}

	routineCount := len(*ckeNodes)
	allSuccess := true
	nodesResults := make([]NodeResult, 0)
	for {
		select {
		case res := <-ch:
			routineCount--
			if res.nodeName != "" {
				allSuccess = false
				nodesResults = append(nodesResults, res)
			}
			if routineCount == 0 {
				RemoveComponentScripts()
				if !allSuccess {
					for _, r := range nodesResults {
						logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
					}
					return fmt.Errorf("not all nodes are successful, see log for detail")
				} else {
					logrus.Info("All nodes are successful.")
					return nil
				}
			}
		}
	}
}

//execute ckectl delete cmd at one node to process one component
func ExecuteCkectlDelete(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {
	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}

	//	如果是运行时组件，则将删除脚本文件拷贝过去
	if component.Labels["containerRuntime"] == "true" {
		scpCmd := fmt.Sprintf("scp -o StrictHostKeyChecking=no -i %s -r %s %s:%s", keyFile, constants.CkeComponentScriptPath+"/"+component.Name, node.Spec.Address, constants.CkectlScriptPath)
		cmd := exec.Command("sh", "-c", scpCmd)
		output, err := cmd.CombinedOutput()
		logrus.Infof("copy delete-scripts to target node,Node:%s,Component:%s,Output:%s",node.Name,component.Name,string(output))
		if err != nil {
			logrus.Errorf("execute scp cmd failed: %s", err.Error())
			return fmt.Errorf("execute scp cmd failed: %s", err.Error())
		}
	}

	var ckectlCmd = fmt.Sprintf(`%s component delete --component %s --cluster %s --node %s --dir %s  --address %s --port %d --nodeName %s --key %s --suffix %s`,
		constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
	//日志暂未加任何包装
	stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
	if err != nil {
		logrus.Errorf("ckectl component delete failed, err: %s", err.Error())
		return err
	} else {
		// read from the output channel until the done signal is passed
		isTimeout := true
	loop2:
		for {
			select {
			case isTimeout = <-doneChan:
				break loop2
			case outline := <-stdoutChan:
				logrus.Info("ckectlCmd:stdout:" + node.Name + "_" + component.Name + ":", outline)
			case errline := <-stderrChan:
				logrus.Info("ckectlCmd:stderr:" + node.Name + "_" + component.Name + ":", errline)
			case err = <-errChan:
			}
		}

		// get exit code or command error.
		if err != nil {
			fmt.Println("err: " + err.Error())
			return err
		}

		// command time out
		if !isTimeout {
			logrus.Errorf("ckectl component delete failed, err: command timeout")
			return fmt.Errorf("ckectl component delete failed, err: command timeout")
		}
	}
	return nil
}

//容器中额外挂载一个/script
func UpgradeComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig) error {
	//调用对应节点的ckectl命令执行安装
	//docker需要传安装包和脚本
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		//copy script to /opt/script-template
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		for _, component := range *components {
			// #nosec
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}
	//运行时安装 需要直传安装包和脚本
	//确定执行节点，确定节点的角色 调用ckectl
	//cluster级别的直接起pod执行
	type NodeResult struct {
		nodeName      string
		componentName string
	}
	ch := make(chan NodeResult)
	clusterExecuted := false
	suffix := RandString(6)
	for _, node := range *ckeNodes {
		//选择一个master执行cluster操作
		isClusterExecutedNode := false
		if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
			clusterExecuted = true
			isClusterExecutedNode = true
		}
		go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
			keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string) {
			for _, component := range *components {
				logrus.Infof("Upgrading component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/upgrade"
				if !util.Exists(workPath) {
					logrus.Infof("upgrade dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					continue
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				//过滤出componentCr中指定的节点
				installedNodesName := component.Status.Nodes
				//dir: master slave node cluster
				for index, dir := range dirWithoutNumber {
					//获取节点 调用
					if InArray(node.ObjectMeta.Name, installedNodesName) && isCluster && (dir == "cluster") {
						//todo 改造成job形式
						err = ExecuteCkectlUpgrade(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlCreate failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
						continue
					}
					if InArray(node.ObjectMeta.Name, installedNodesName) && NodeIsInDir(dir, node.Spec.Roles) {
						//调用指定节点的ckectl component upgrade
						err = ExecuteCkectlUpgrade(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlCreate failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
					}
				}
			}
			ch <- NodeResult{}
			return
		}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix)
	}

	routineCount := len(*ckeNodes)
	allSuccess := true
	nodesResults := make([]NodeResult, 0)
	for {
		select {
		case res := <-ch:
			routineCount--
			if res.nodeName != "" {
				allSuccess = false
				nodesResults = append(nodesResults, res)
			}
			if routineCount == 0 {
				RemoveComponentScripts()
				if !allSuccess {
					for _, r := range nodesResults {
						logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
					}
					return fmt.Errorf("not all nodes are successful, see log for detail")
				} else {
					logrus.Info("All nodes are successful.")
					return nil
				}
			}
		}
	}
}

//execute ckectl cmd at one node to process one component
func ExecuteCkectlUpgrade(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {
	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}

	if component.Labels["containerRuntime"] == "true" {
		var mkdirCmd = fmt.Sprintf("mkdir -p %s", constants.CkectlScriptPath)
		_, _, _, err := util.SetupSSH(node.Spec.Address, keyFile).Run(mkdirCmd, 60*time.Second)
		if err != nil {
			logrus.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
			return fmt.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
		}

		scpCmd := fmt.Sprintf("scp -o StrictHostKeyChecking=no -i %s -r %s %s:%s", keyFile, constants.CkeComponentScriptPath+"/"+component.Name, node.Spec.Address, constants.CkectlScriptPath)
		// #nosec
		cmd := exec.Command("sh", "-c", scpCmd)
		output, err := cmd.CombinedOutput()
		logrus.Infof("scp output: %s", string(output))
		if err != nil {
			logrus.Errorf("execute scp cmd failed: %s", err.Error())
			return fmt.Errorf("execute scp cmd failed: %s", err.Error())
		}
		var ckectlCmd = fmt.Sprintf(`%s component upgrade --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component upgrade failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop1:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop1
				case outline := <-stdoutChan:
					logrus.Info("out:", outline)
				case errline := <-stderrChan:
					logrus.Info("out:", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl component upgrade failed, err: command timeout")
				return fmt.Errorf("ckectl component upgrade failed, err: command timeout")
			}
		}

		return nil
	} else {
		var ckectlCmd = fmt.Sprintf(`%s component upgrade --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component upgrade failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop2:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop2
				case outline := <-stdoutChan:
					logrus.Info("out:", outline)
				case errline := <-stderrChan:
					logrus.Info("out:", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl upgrade create failed, err: command timeout")
				return fmt.Errorf("ckectl component upgrade failed, err: command timeout")
			}
		}
		return nil
	}
}

//容器中额外挂载一个/script
func UpdateComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig) error {
	//调用对应节点的ckectl命令执行安装
	//docker需要传安装包和脚本
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		//copy script to /opt/script-template
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		for _, component := range *components {
			// #nosec
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}
	//运行时安装 需要直传安装包和脚本
	//确定执行节点，确定节点的角色 调用ckectl
	//cluster级别的直接起pod执行
	type NodeResult struct {
		nodeName      string
		componentName string
	}
	ch := make(chan NodeResult)
	clusterExecuted := false
	suffix := RandString(6)
	for _, node := range *ckeNodes {
		//选择一个master执行cluster操作
		isClusterExecutedNode := false
		if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
			clusterExecuted = true
			isClusterExecutedNode = true
		}
		go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
			keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string) {
			for _, component := range *components {
				logrus.Infof("Updating component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/update"
				if !util.Exists(workPath) {
					logrus.Infof("update dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					continue
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				//过滤出componentCr中指定的节点
				installedNodesName := component.Status.Nodes

				//dir: master slave node cluster
				for index, dir := range dirWithoutNumber {
					//获取节点 调用
					if InArray(node.ObjectMeta.Name, installedNodesName) && isCluster && (dir == "cluster") {
						//todo 改造成job形式
						err = ExecuteCkectlUpdate(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlCreate failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
						continue
					}
					if InArray(node.ObjectMeta.Name, installedNodesName) && NodeIsInDir(dir, node.Spec.Roles) {
						//调用指定节点的ckectl component upgrade
						err = ExecuteCkectlUpgrade(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlCreate failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
					}
				}
			}
			ch <- NodeResult{}
			return
		}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix)
	}

	routineCount := len(*ckeNodes)
	allSuccess := true
	nodesResults := make([]NodeResult, 0)
	for {
		select {
		case res := <-ch:
			routineCount--
			if res.nodeName != "" {
				allSuccess = false
				nodesResults = append(nodesResults, res)
			}
			if routineCount == 0 {
				RemoveComponentScripts()
				if !allSuccess {
					for _, r := range nodesResults {
						logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
					}
					return fmt.Errorf("not all nodes are successful, see log for detail")
				} else {
					logrus.Info("All nodes are successful.")
					return nil
				}
			}
		}
	}
}

//execute ckectl cmd at one node to process one component
func ExecuteCkectlUpdate(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {
	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}

	if component.Labels["containerRuntime"] == "true" {
		var mkdirCmd = fmt.Sprintf("mkdir -p %s", constants.CkectlScriptPath)
		_, _, _, err := util.SetupSSH(node.Spec.Address, keyFile).Run(mkdirCmd, 60*time.Second)
		if err != nil {
			logrus.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
			return fmt.Errorf("make ckectl-script dir at %s failed: %s", node.Name, err.Error())
		}

		scpCmd := fmt.Sprintf("scp -o StrictHostKeyChecking=no -i %s -r %s %s:%s", keyFile, constants.CkeComponentScriptPath+"/"+component.Name, node.Spec.Address, constants.CkectlScriptPath)
		// #nosec
		cmd := exec.Command("sh", "-c", scpCmd)
		output, err := cmd.CombinedOutput()
		logrus.Infof("scp output: %s", string(output))
		if err != nil {
			logrus.Errorf("execute scp cmd failed: %s", err.Error())
			return fmt.Errorf("execute scp cmd failed: %s", err.Error())
		}

		var ckectlCmd = fmt.Sprintf(`%s component update --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component update failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop1:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop1
				case outline := <-stdoutChan:
					logrus.Info("out:", outline)
				case errline := <-stderrChan:
					logrus.Info("out:", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl component update failed, err: command timeout")
				return fmt.Errorf("ckectl component update failed, err: command timeout")
			}
		}

		return nil
	} else {
		var ckectlCmd = fmt.Sprintf(`%s component update --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
			constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
		//日志暂未加任何包装
		stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
		if err != nil {
			logrus.Errorf("ckectl component update failed, err: %s", err.Error())
			return err
		} else {
			// read from the output channel until the done signal is passed
			isTimeout := true
		loop2:
			for {
				select {
				case isTimeout = <-doneChan:
					break loop2
				case outline := <-stdoutChan:
					logrus.Info("out:", outline)
				case errline := <-stderrChan:
					logrus.Info("out:", errline)
				case err = <-errChan:
				}
			}

			// get exit code or command error.
			if err != nil {
				fmt.Println("err: " + err.Error())
				return err
			}

			// command time out
			if !isTimeout {
				logrus.Errorf("ckectl component update failed, err: command timeout")
				return fmt.Errorf("ckectl component update failed, err: command timeout")
			}
		}

		return nil
	}
}

/*
//容器中额外挂载一个/script
func Master2SlaveComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig) error {
	//调用对应节点的ckectl命令执行安装
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		//copy script to /opt/script-template
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		for _, component := range *components {
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}
	//发送给组件的全部已安装节点执行
	//cluster级别的直接起pod执行
	//不需要额外计算目标节点
	type NodeResult struct {
		nodeName      string
		componentName string
	}
	ch := make(chan NodeResult)
	clusterExecuted := false
	suffix := RandString(6)
	for _, node := range *ckeNodes {
		//选择一个master执行cluster操作
		isClusterExecutedNode := false
		if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
			clusterExecuted = true
			isClusterExecutedNode = true
		}
		go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
			keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string) {
			for _, component := range *components {
				logrus.Infof("Master2Slave component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/master2slave"
				if !util.Exists(workPath) {
					logrus.Infof("master2slave dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					continue
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				installedNodesName := component.Status.Nodes
				match := false
				if InArray(node.ObjectMeta.Name, installedNodesName) || InArray("master2slave", node.Spec.Roles) {
					//若节点在已经安装的节点中或节点标记为了master2slave节点
					match = true
				}

				//dir: master slave node cluster
				//获取节点类别的执行顺序
				for index, dir := range dirWithoutNumber {
					//获取节点 调用
					if isCluster && (dir == "cluster") {
						//todo 改造成job形式
						err = ExecuteCkectlMaster2Slave(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlMaster2Slave failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
						continue
					}
					if match && NodeIsInDir(dir, node.Spec.Roles) {
						//调用指定节点的ckectl component scaleup
						err = ExecuteCkectlMaster2Slave(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlMaster2Slave failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
					}
				}
			}
			ch <- NodeResult{}
			return
		}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix)
	}

	routineCount := len(*ckeNodes)
	allSuccess := true
	nodesResults := make([]NodeResult, 0)
	for {
		select {
		case res := <-ch:
			routineCount--
			if res.nodeName != "" {
				allSuccess = false
				nodesResults = append(nodesResults, res)
			}
			if routineCount == 0 {
				RemoveComponentScripts()
				if !allSuccess {
					for _, r := range nodesResults {
						logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
					}
					return fmt.Errorf("not all nodes are successful, see log for detail")
				} else {
					logrus.Info("All nodes are successful.")
					return nil
				}
			}
		}
	}
}

//execute ckectl master2slave cmd at one node to process one component
func ExecuteCkectlMaster2Slave(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {
	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}

	var ckectlCmd = fmt.Sprintf(`%s component master2slave --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
		constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
	//日志暂未加任何包装
	stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
	if err != nil {
		logrus.Errorf("ckectl component master2slave failed, err: %s", err.Error())
		return err
	} else {
		// read from the output channel until the done signal is passed
		isTimeout := true
	loop2:
		for {
			select {
			case isTimeout = <-doneChan:
				break loop2
			case outline := <-stdoutChan:
				logrus.Info("out:", outline)
			case errline := <-stderrChan:
				logrus.Info("out:", errline)
			case err = <-errChan:
			}
		}

		// get exit code or command error.
		if err != nil {
			fmt.Println("err: " + err.Error())
			return err
		}

		// command time out
		if !isTimeout {
			logrus.Errorf("ckectl component master2slave failed, err: command timeout")
			return fmt.Errorf("ckectl component master2slave failed, err: command timeout")
		}
	}

	return nil
}

//容器中额外挂载一个/script
func Slave2MasterComponent(components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster, ckeNodes *[]ckeadmv1beta1.CkeNode, keyFile string, watchConfig ckeadm.WatchConfig) error {
	//调用对应节点的ckectl命令执行安装
	//若容器中运行 则不能使用docker run
	inContainer := true
	_, err := os.Stat(constants.CkeComponentScriptTempRoot)
	if err != nil {
		if os.IsNotExist(err) {
			inContainer = false
		} else {
			return fmt.Errorf("failed to get inContainer status, %s", err.Error())
		}
	}

	//sort ckeNode
	sort.Sort(util.CkeNodeSort(*ckeNodes))

	//get master address list
	addressArray := make([]string, 0)
	for _, node := range *ckeNodes {
		if InArray("master", node.Spec.Roles) {
			addressArray = append(addressArray, node.Spec.Address)
			if !InArray("worker", node.Spec.Roles) {
				node.Spec.Roles = append(node.Spec.Roles, "worker")
			}
		}
	}
	addressList := strings.Replace(strings.Trim(fmt.Sprint(addressArray), "[]"), " ", ",", -1)

	if !inContainer {
		//ckeadm-wait log service
		ckeadmwait := logscmd.NewCkeadmWaitService(constants.CkeadmWatchServiceName, constants.WhichCkeadm, watchConfig.ListenAddress,
			watchConfig.ListenPort, watchConfig.LogCacheDir, false, keyFile, addressList)
		if ckeadmwait.IsRunning() == false {
			if err := ckeadmwait.CreateAndStartService(logscmd.CkeadmWaitServiceTemplate, os.Args[0]); err != nil {
				return fmt.Errorf("error while set up ckeadm wait service: %s", err.Error())
			}
		}

		//config docker
		docker := runner.NewDockerService("docker", constants.DockerdPath, runner.GetDockerCgroupDriver(),
			runner.GetRegistry(cluster.Spec.Registry.Domain, cluster.Spec.Registry.Port))
		if docker.CheckServiceConfig() == false {
			if err := docker.CreateAndStartService(runner.DockerServiceTemplate, constants.WhichDockerd); err != nil {
				return fmt.Errorf("error while set up docker service: %s", err.Error())
			}
		}

		//copy script template
		for _, component := range *components {
			containerName := component.ObjectMeta.Name + "-template"
			err = ExtractScript(containerName, GetImageName(cluster, component), component)
			if err != nil {
				return fmt.Errorf("error while extract script from template image: %s", err.Error())
			}
		}
	}

	if inContainer {
		//copy script to /opt/script-template
		_ = os.Mkdir(constants.CkeComponentScriptPath, 0644)
		for _, component := range *components {
			var copyScript = exec.Command("/bin/cp", "-rf", constants.CkeComponentScriptTempRoot+"/"+component.ObjectMeta.Name, constants.CkeComponentScriptPath)
			err = copyScript.Run()
			if err != nil {
				return fmt.Errorf("error while copy script to %s: %s", constants.CkeComponentScriptPath, err.Error())
			}
		}
	}
	//发送给组件的全部已安装节点执行
	//cluster级别的直接起pod执行
	//不需要额外计算目标节点
	type NodeResult struct {
		nodeName      string
		componentName string
	}
	ch := make(chan NodeResult)
	clusterExecuted := false
	suffix := RandString(6)
	for _, node := range *ckeNodes {
		//选择一个master执行cluster操作
		isClusterExecutedNode := false
		if (!clusterExecuted) && InArray("master", node.Spec.Roles) {
			clusterExecuted = true
			isClusterExecutedNode = true
		}
		go func(isCluster bool, node ckeadmv1beta1.CkeNode, components *[]ckeadmv1beta1.CkeComponent, cluster ckeadmv1beta1.CkeCluster,
			keyFile string, watchConfig ckeadm.WatchConfig, allNodes *[]ckeadmv1beta1.CkeNode, ch chan NodeResult, suffix string) {
			for _, component := range *components {
				logrus.Infof("Slave2Master component %s on node %s ...", component.ObjectMeta.Name, node.ObjectMeta.Name)
				workPath := constants.CkeComponentScriptPath + "/" + component.ObjectMeta.Name + "/slave2master"
				if !util.Exists(workPath) {
					logrus.Infof("slave2master dir not exists in component %s template, skipped...", component.ObjectMeta.Name)
					continue
				}
				dirs, err := util.ListSubDirsInPath(workPath)
				if err != nil {
					logrus.Errorf("error while get dirs from template: %s, node routine: %s", err.Error(), node.ObjectMeta.Name)
					ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
					return
				}
				sort.Sort(sort.StringSlice(dirs))

				//去除序号
				dirWithoutNumber := make([]string, 0)
				for _, dir := range dirs {
					reg := regexp.MustCompile(`^[0-9]+-`)
					dirWithoutNumber = append(dirWithoutNumber, reg.ReplaceAllString(dir, ""))
				}

				installedNodesName := component.Status.Nodes
				match := false
				if InArray(node.ObjectMeta.Name, installedNodesName) || InArray("slave2master", node.Spec.Roles) {
					//若节点在已经安装的节点中或节点标记为了slave2master节点
					match = true
				}

				//dir: master slave node cluster
				//获取节点类别的执行顺序
				for index, dir := range dirWithoutNumber {
					//获取节点 调用
					if isCluster && (dir == "cluster") {
						//todo 改造成job形式
						err = ExecuteCkectlSlave2Master(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlSlave2Master failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
						continue
					}
					if match && NodeIsInDir(dir, node.Spec.Roles) {
						//调用指定节点的ckectl component scaleup
						err = ExecuteCkectlSlave2Master(component, node, cluster, ckeNodes, keyFile, dirs[index], watchConfig.ListenAddress, watchConfig.ListenPort, suffix)
						if err != nil {
							logrus.Errorf("ExecuteCkectlSlave2Master failed at node %s: %s", node.ObjectMeta.Name, err.Error())
							ch <- NodeResult{nodeName: node.Name, componentName: component.Name}
							return
						}
					}
				}
			}
			ch <- NodeResult{}
			return
		}(isClusterExecutedNode, node, components, cluster, keyFile, watchConfig, ckeNodes, ch, suffix)
	}

	routineCount := len(*ckeNodes)
	allSuccess := true
	nodesResults := make([]NodeResult, 0)
	for {
		select {
		case res := <-ch:
			routineCount--
			if res.nodeName != "" {
				allSuccess = false
				nodesResults = append(nodesResults, res)
			}
			if routineCount == 0 {
				RemoveComponentScripts()
				if !allSuccess {
					for _, r := range nodesResults {
						logrus.Errorf("Component %s failed in node: %s, see log for detail.", r.componentName, r.nodeName)
					}
					return fmt.Errorf("not all nodes are successful, see log for detail")
				} else {
					logrus.Info("All nodes are successful.")
					return nil
				}
			}
		}
	}
}

//execute ckectl slave2master cmd at one node to process one component
func ExecuteCkectlSlave2Master(component ckeadmv1beta1.CkeComponent, node ckeadmv1beta1.CkeNode, cluster ckeadmv1beta1.CkeCluster,
	allNodes *[]ckeadmv1beta1.CkeNode, keyFile, dir, address string, port int, suffix string) error {
	componentByte, err := runner.ComponentToYaml(component)
	if err != nil {
		return fmt.Errorf("error marshal CkeComponent into bytes: %s", err.Error())
	}
	componentCode := base64.StdEncoding.EncodeToString(componentByte)

	nodeByte, err := runner.NodesToYaml(allNodes)
	if err != nil {
		return fmt.Errorf("error marshal CkeNode into bytes: %s", err.Error())
	}
	nodeCode := base64.StdEncoding.EncodeToString(nodeByte)

	clusterByte, err := runner.ClusterToYaml(cluster)
	if err != nil {
		return fmt.Errorf("error marshal CkeCluster into bytes: %s", err.Error())
	}
	clusterCode := base64.StdEncoding.EncodeToString(clusterByte)

	keyCode, err := runner.GetPrivateKeyFromFile(keyFile)
	if err != nil {
		logrus.Errorf("get key failed: %s", err.Error())
		return fmt.Errorf("get key failed: %s", err.Error())
	}

	var ckectlCmd = fmt.Sprintf(`%s component slave2master --component %s --cluster %s --node %s --dir %s --address %s --port %d --nodeName %s --key %s --suffix %s`,
		constants.WhichCkectl, componentCode, clusterCode, nodeCode, dir, address, port, node.Name, keyCode, suffix)
	//日志暂未加任何包装
	stdoutChan, stderrChan, doneChan, errChan, err := util.SetupSSH(node.Spec.Address, keyFile).Stream(ckectlCmd, 30*time.Minute)
	if err != nil {
		logrus.Errorf("ckectl component slave2master failed, err: %s", err.Error())
		return err
	} else {
		// read from the output channel until the done signal is passed
		isTimeout := true
	loop2:
		for {
			select {
			case isTimeout = <-doneChan:
				break loop2
			case outline := <-stdoutChan:
				logrus.Info("out:", outline)
			case errline := <-stderrChan:
				logrus.Info("out:", errline)
			case err = <-errChan:
			}
		}

		// get exit code or command error.
		if err != nil {
			fmt.Println("err: " + err.Error())
			return err
		}

		// command time out
		if !isTimeout {
			logrus.Errorf("ckectl component slave2master failed, err: command timeout")
			return fmt.Errorf("ckectl component slave2master failed, err: command timeout")
		}
	}

	return nil
}
*/

func RemoveComponentScripts() {
	_ = os.RemoveAll(constants.CkeComponentScriptPath)
}

// RandString 生成随机字符串
func RandString(len int) string {
	var r *rand.Rand
	r = rand.New(rand.NewSource(time.Now().Unix()))
	bytes := make([]byte, len)
	for i := 0; i < len; i++ {
		b := r.Intn(26) + 97
		bytes[i] = byte(b)
	}
	return string(bytes)
}

//clear create.log and record begin time
func RecordBeginTime(cluster string) {
	file := fmt.Sprintf("%s/ckeadm-status/%s/begin.log", constants.CkeComponentLogPath, cluster)
	//remove old dirs and files
	_ = os.RemoveAll(path.Dir(file))
	//make new dir
	err := os.MkdirAll(path.Dir(file), os.ModePerm)
	if err != nil {
		logrus.Errorf("MkdirAll %s failed: %s", path.Dir(file), err.Error())
	}
	writer, err := os.OpenFile(filepath.Clean(file), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
	if err != nil {
		logrus.Errorf("Create ckeadm wait log file failed: %s", err.Error())
		return
	}
	defer writer.Close()
	_, err = fmt.Fprintln(writer, "BeginTime#"+time.Now().Format(time.RFC3339))
	if err != nil {
		logrus.Errorf("Record begin time failed: %s", err.Error())
		return
	}
}

func RecordEndTime(cluster string) {
	file := fmt.Sprintf("%s/ckeadm-status/%s/end.log", constants.CkeComponentLogPath, cluster)
	writer, err := os.OpenFile(filepath.Clean(file), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
	if err != nil {
		logrus.Errorf("Open ckeadm wait log file failed: %s", err.Error())
		return
	}
	defer writer.Close()
	_, err = fmt.Fprintln(writer, "EndTime#"+time.Now().Format(time.RFC3339))
	if err != nil {
		logrus.Errorf("Record end time failed: %s", err.Error())
		return
	}
}

//progress -1 means do not update progress
func RecordStatus(cluster, node, status string, progress int, components []string) {
	file := fmt.Sprintf("%s/ckeadm-status/%s/%s.log", constants.CkeComponentLogPath, cluster, node)

	if !util.Exists(file) {
		f, err := os.Create(filepath.Clean(file))
		if err != nil {
			_ = f.Close()
		}
	}

	//content, err := ioutil.ReadFile(file)
	//if err == nil {
	//	fmt.Println("CKE install status: \n", string(content))
	//}
	reader, err := os.Open(filepath.Clean(file))
	if err != nil {
		logrus.Errorf("Read ckeadm wait log file failed: %s", err.Error())
		return
	}

	defer reader.Close()

	//file may contain or not contain progress of one node
	//if exists, update; if not, append.
	lineArray := make([]string, 0)
	rd := bufio.NewReader(reader)
	updated := false
	for {
		line, err := rd.ReadString('\n')
		if err != nil || io.EOF == err {
			break
		}
		if strings.Contains(line, node) {
			//update line
			if progress == -1 {
				progress, err = strconv.Atoi(strings.ReplaceAll(strings.Split(line, "#")[2], "%", ""))
				if err != nil {
					logrus.Errorf("Convert progress to int failed: %s", err.Error())
				}
			}
			line = makeStatusLine(node, status, progress, components)
			updated = true
		}
		lineArray = append(lineArray, line)
	}
	if !updated {
		//append new line
		newLine := makeStatusLine(node, status, progress, components)
		lineArray = append(lineArray, newLine)
	}
	//write lineArray to file
	writer, err := os.OpenFile(filepath.Clean(file), os.O_WRONLY|os.O_TRUNC, 0644)
	if err != nil {
		logrus.Errorf("Open ckeadm wait log file failed: %s", err.Error())
		return
	}
	defer writer.Close()

	for _, line := range lineArray {
		line = strings.ReplaceAll(line, "\n", "")
		if line == "" {
			continue
		}
		_, err = fmt.Fprintln(writer, line)
		if err != nil {
			logrus.Errorf("Write ckeadm wait log file failed: %s", err.Error())
		}
	}

	//content, err = ioutil.ReadFile(file)
	//if err == nil {
	//	fmt.Println("CKE install status refreshed: \n", string(content))
	//}
}

func makeStatusLine(node, status string, progress int, components []string) string {
	message := ""
	if status == constants.CkeStatusCompleted {
		message = "Done"
	} else if status == constants.CkeStatusPrepare {
		message = "Preparing to install node ..."
	} else {
		message = fmt.Sprintf("%s: %s", status, strings.Join(components, ", "))
	}
	return fmt.Sprintf("%s#%s#%d%%#%s\n", node, status, progress, message)
}

func getProgress(list [][]string, index int) int {
	processed := 0
	total := 0
	for k, v := range list {
		if k < index {
			processed += len(v)
		}
		total += len(v)
	}
	return (processed + 1) * 100 / (total + 1)
}
